package com.gking.centerSystem.utils;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.function.Consumer;


/**
 * 设置好的多个 SSE 服务
 */
public class SseServer {

    /**
     * 聊天 SSE 服务
     */
    private static final SseGenerator chattingSseServer = new SseGenerator();

    //连接 聊天 SSE
    public static SseEmitter chattingConnectSSE(String userId) {
        return chattingSseServer.connect(userId);
    }

    //发送聊天对话消息
    public static <T> boolean chattingSendMessage(String receiverId, T message) {
        return chattingSseServer.sendMessage(receiverId, message);
    }

    //发送聊天指定事件
    private static <T> boolean chattingSendTargetEvent(String receiverId, String eventName, T data) {
        return chattingSseServer.sendTargetEvent(receiverId, eventName, data);
    }

    //批量发送未读消息
//    public static boolean sendNotReadRecordList(String receiverId, List<ChatHistory> historyList) {
//        return chattingSendTargetEvent(receiverId, "notReadRecordList", historyList);
//    }

    /**
     * 持久 SSE 服务
     */
    private static final SseGenerator enduringSseServer = new SseGenerator();

    //连接 持久消息通知 SSE
    public static SseEmitter enduringConnectSSE(String userId) {
        return enduringSseServer.connect(userId);
    }

    //发送通知消息
    public static <T> boolean enduringSendMessage(String receiverId, T message) {
        return enduringSseServer.sendMessage(receiverId, message);
    }

    //发送通知指定事件
    private static <T> boolean enduringSendTargetEvent(String receiverId, String eventName, T data) {
        return enduringSseServer.sendTargetEvent(receiverId, eventName, data);
    }

    public static boolean notReadCountEvent(String userId, Integer count) {
        return enduringSendTargetEvent(userId, "notReadCount", count);
    }

    public static boolean refreshTokenEvent(String userId, String token) {
        return enduringSendTargetEvent(userId, "refreshToken", token);
    }

}


/**
 * SSE 构造器
 */
@Slf4j
class SseGenerator {
    private final Map<String, SseEmitter> sseEmitterMap = new ConcurrentHashMap<>();

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    /**
     * sse 连接
     *
     * @param userId 用户id
     * @return s
     */
    public SseEmitter connect(String userId) {
        // 设置超时时间，0表示不过期。默认30秒
        SseEmitter sseEmitter = new SseEmitter(30000L);

        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));     // 长链接完成后回调接口(即关闭连接时调用)
        sseEmitter.onTimeout(timeoutCallBack(userId));        // 连接超时回调
        sseEmitter.onError(errorCallBack(userId));          // 推送消息异常时，回调方法

        sseEmitterMap.put(userId, sseEmitter);

        sendMessage(userId, "连接成功！");

        log.info("SSE 现在在线人数为：{}", getOnlineCount());
        log.info("SSE 在线人数详情：{}", sseEmitterMap);

        return sseEmitter;
    }


    /**
     * 发送消息给用户
     *
     * @param receiverId 目标用户id
     * @param message    消息详情
     * @return s
     */
    public <T> boolean sendMessage(String receiverId, T message) {
        if (sseEmitterMap.containsKey(receiverId)) {
            try {
                log.info("SSE 向用户 {} 推送消息： {} ", receiverId, message);
                sseEmitterMap.get(receiverId).send(message);
                return true;
            } catch (IOException e) {
                log.error("SSE 用户 {} 推送异常: {}", receiverId, e.getMessage());
            }
        } else {
            executorService.submit(() -> {
                try {
                    for (int i = 0; i < 5; i++) {
                        Thread.sleep(400);
                        if (!sseEmitterMap.containsKey(receiverId)) continue;
                        sendMessage(receiverId, message);
                        break;
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        return false;
    }


    /**
     * 向特定用户发起特定事件数据
     *
     * @param receiverId 目标用户id
     * @param eventName  事件名称
     * @param data       数据
     */
    public <T> boolean sendTargetEvent(String receiverId, String eventName, T data) {
        if (sseEmitterMap.containsKey(receiverId)) {
            log.info("SSE 对用户 {} 发起事件 {} ， data为 : {}", receiverId, eventName, data);
            try {
                SseEmitter.SseEventBuilder builder = SseEmitter.event()
                        .name(eventName)
                        .id(UUID.randomUUID().toString())
                        .data(data);

                sseEmitterMap.get(receiverId).send(builder);
                return true;
            } catch (IOException e) {
                log.error("SSE 用户 {} 推送 {} 事件异常: 数据为 {}", receiverId, eventName, e.getMessage());
            }
        } else {
            try {
                for (int i = 0; i < 5; i++) {
                    Thread.sleep(400);
                    if (!sseEmitterMap.containsKey(receiverId)) continue;
                    sendTargetEvent(receiverId, eventName, data);
                    break;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return false;
    }


    //获取在线人数
    private synchronized int getOnlineCount() {
        return sseEmitterMap.size();
    }


    //移除用户连接
    private void removeUser(String userId) {
        if (sseEmitterMap.containsKey(userId)) {
            sseEmitterMap.remove(userId);
            log.info("SSE 在线连接数 {}， 连接详情： {} ", getOnlineCount(), sseEmitterMap.toString());
        }
    }

    //长链接完成后回调接口(即关闭连接时调用)
    private Runnable completionCallBack(String userId) {
        return () -> {
//            log.info("SSE 连接结束：{}", userId);
            removeUser(userId);
        };
    }

    //连接超时时调用
    private Runnable timeoutCallBack(String userId) {
        return () -> {
//            log.warn("SSE 连接超时： {}", userId);
            removeUser(userId);
        };
    }

    //推送消息异常时，回调方法
    private Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            log.error("SSE sseEmitter [ errorCallBack ]  : 连接异常，客户端 id: {}", userId);
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(500);
                    SseEmitter sseEmitter = sseEmitterMap.get(userId);
                    if (sseEmitter == null) {
                        log.error("SSE sseEmitter [ errorCallBack ] : 第 {} 次消息重推失败，客户端 id: {}", i + 1, userId);
                        continue;
                    }
                    sseEmitter.send("失败后重新推送");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }
}


